package edu.kit.aifb.ldstream.repository.event;

import java.io.IOException;
import java.io.Reader;
import java.io.StringReader;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Semaphore;

import org.openrdf.model.impl.URIImpl;
import org.openrdf.repository.RepositoryConnection;
import org.openrdf.repository.RepositoryException;
import org.openrdf.rio.RDFFormat;
import org.openrdf.rio.RDFParseException;

import edu.kit.aifb.ldstream.env.Environment;
import edu.kit.aifb.ldstream.repository.listeners.AbstractListeningThread;

public class EventWorker extends Thread{

	private ConcurrentLinkedQueue<StreamEvent> queue;
	private RepositoryConnection connection;
	private Reader reader;
	private Semaphore semaphore;
	private boolean finished = false;

	public EventWorker(ConcurrentLinkedQueue<StreamEvent> queue, Semaphore semaphore){
		this.queue = queue;
		this.semaphore = semaphore;
	}

	public void run(){
		try {
			this.connection = Environment.REPO.getConnection();
			this.connection.setAutoCommit(false);
		} catch (RepositoryException e) {
			e.printStackTrace();
		}

		StreamEvent currentEvent = null;

		while(!this.finished){

			try {
				// the semaphore is used because of the inaccurate size method of the queue.
				this.semaphore.acquire();
			} catch (InterruptedException e2) {
				e2.printStackTrace();
			}

			if(finished){
				break;
			}
			currentEvent = this.queue.poll();
			// check every format
			for(RDFFormat currentFormat : Environment.formats){
				this.reader = new StringReader(currentEvent.getEvent());
				try {
					this.connection.add(this.reader, currentEvent.getBaseURI(), currentFormat, 
							new URIImpl(Environment.EVENT_CONTEXT + AbstractListeningThread.EVENT_ID.getAndAdd(1) + ">"));
					if(Environment.VERBOSE){
						System.out.println("Successfully saved " + currentEvent.getURI() + "to repository");
					}
					// if we reach this point we were able to fill the repository.
					break;

				} catch (RDFParseException e) {
					try {
						this.reader.reset();
					} catch (IOException e1) {
						e1.printStackTrace();
					}
				} catch (RepositoryException e) {
					e.printStackTrace();
				} catch (IOException e) {
					e.printStackTrace();
				}
				try {
					this.reader.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		shutdown();
	}

	public void isFinished(){
		this.finished = true;
	}

	private void shutdown(){	
		if(this.reader != null){
			try {
				this.reader.close();
			} catch (IOException e1) {
				e1.printStackTrace();
			}
		}
		try {
			this.connection.commit();
			this.connection.close();
		} catch (RepositoryException e) {
			e.printStackTrace();
		}
	}
}
